<?php
abstract class BaseQueueProcessor
{
	protected $data;
	protected $conf;

	protected $other_queue = array();
	protected $cache;
	// 正在跑的队列名
	protected $queue_name;

	//$data是原始数据，如果不需要用可以直接使用$this->data是已解释过的
	abstract public function parse();

	// 用来标识一个任务的处理时间
	protected $start_time;
	protected $end_time;
	protected $p_start_time;//进程开始时间
	protected $error_msg='';
	protected $mobile_type = 0;
	protected $fail_in=false;//失败了继续放入队列
	protected $fail_in_time=10;//失败后sleep 多久放入队列

	// 渠道相关信息
	protected $channel_id = 'hequan';    // 1 天猫
	protected $appkey = "";
	protected $secretkey = "";
	protected $sessionkey = "";
    protected $release = false;
	/**
	 * 主队列
	 * @var Queue
	 */
	private $_queue;

	private static $running;

	public function __construct($name, $conf)
	{
		$this->setSignalHandle ();
		$this->_queue = new Queue ( 'w', $name );
		$this->queue_name = $name;
		$this->conf = $conf;

		QueueLog::setQueueName($name);
		self::$running = true;
		$this->p_start_time=time();
	}

	public function setData($data)
	{
		$this->data = json_decode ( $data, true );
	}

	public function setSignalHandle() {
		declare ( ticks = 1 );
		pcntl_signal ( SIGINT, array ($this, 'sighandler' ) );
		pcntl_signal ( SIGTERM, array ($this, 'sighandler' ) );
	}

	static function sighandler($signo) {
		QueueLog::debugLog ( " script is stopped" );
		self::$running = false;
	}

	private function getMsg($at_head = false)
	{
		// 获取任务的起始时间
		$this->start_time = time();

		$ret = false;
		if (! $this->_queue) {
			die ( "未分配队列" );
		}

		try
		{
			$ret = $this->_queue->pget ( $at_head );
		} catch ( Exception $ex ) {
			//$this->log->error_log();
			var_dump ( $ex );
		}

		if ($ret === false) {
			trigger_error( $this->queue_name . " get queue msg failed", E_USER_ERROR);
		}

		if ($ret == 'HTTPSQS_GET_END') {
			$ret = false;
		}

		return $ret;
	}
	//队列重启的时候，如果在sleep 就重新放入队列头部
	public function addInQueue(){
	    try
		{
			$data=json_encode($this->data);
			$ret = $this->_queue->pappend($data);
		} catch ( Exception $ex ) {
			$this->errorLog($ex);
		}
	}
	/**
	 * 失败通用处理逻辑
	 * Enter description here ...
	 */
	public function doFail()
	{
		return true;
		//第一次失败发，重试失败不发
		if(empty($this->data['retry_times'])){
			$msg="队列处理失败,queue_name=".$this->queue_name.";".$this->error_msg.';data='.var_export($this->data,true);
		    $data= array( 'mail'=>'hequan@tianpin.com',
						              'msg'=>$msg,
			);
			$data['type']='que';
			$client = MailClient::getInstance();
			@$statusCode = $client->sendSysExMsg($data);
			$this->debugLog($statusCode);
		}
	}
	public function process($count = 0)
	{
		if ($count < 0) {
			return false;
		}

		//trigger_error( $this->queue_name . " is running..." );
		while ( self::$running )
		{
			$at_head = false;

			while ( self::$running && $data = $this->getMsg ( $at_head ))
			{
				$this->setData ( $data );
                //如果传入的释放队列为当前队列 则release为true
				if(isset($this->data['data']['callback']) && $this->data['data']['callback']['release']==$this->queue_name) {
				    $this->release = true;   
				}
				$ret = $this->parse ();

				if ($ret) {
					//处理成功逻辑
					// trigger_error($this->queue_name . ' parse one data succ !');
				} else {
					//处理失败逻辑
					//QueueLog::errorLog('[process failed]:'.$data , $this->queue_name.'_error_'.date('Y_m_d').'.log');
					$this->doFail();
				}
                if(isset($this->data['data']['callback']) && is_array($this->data['data']['callback']) && $this->release) {                        
                    call_user_func_array( array( $this->data['data']['callback']['className'] , $this->data['data']['callback']['method'] ) , $this->data['data']['callback']['params']);
                }
				Log::save();
				$count --;
				if ($count < 1) {
					trigger_error( $this->queue_name . " execute max count" );
					return true;
				}
			}
			$now_houre=date('H',time());
			$start_houre=date('H',$this->p_start_time);//保证每小时重启一次进程

			if (! self::$running || $now_houre!=$start_houre) {
				break;
			}
			//break;
			//QueueLog::debugLog ( "wait 1 second for the next msg " );
			sleep ( 1 );
		}
		trigger_error( $this->queue_name . " quit process" );
	}
    
    /**
     * 设置当前队列是否释放锁
     * @param $acts 动作 如果动作为空或当前动作在动作范围内 则释放该动作的锁
     */
    function setRelease($acts=array()) {
        if(isset($this->data['data']['callback']) && (empty($acts) || in_array($this->data['data']['callback']['params'][0], $acts))) {
            $this->release = true;
        }
    }
	

	/**
	 * 分发到其他队列
	 * @param unknown_type $queues
	 * @param unknown_type $data
	 */
	public function dispatch($queues,$data = null)
	{
		$queues = ( array ) $queues;

		if (empty($data)) {
			$data = $this->data;
		}elseif(isset($this->data['data']['callback'])) {
            $data['data']['callback'] = $this->data['data']['callback'];  
        }
        if(isset($data['data']['callback']) && $data['data']['callback']['release']==$this->queue_name) {
            unset($data['data']['callback']);//传入释放队列未当前队列 则不需要再往下传
        }
		foreach ($queues as $queue_name)
		{
			if(empty($queue_name))
		        continue;
		     do_queue($queue_name,$data['op'],$data['data']);
		}
	}


	public function debugLog($str){
		QueueLog::debugLog($str);
	}

	public function businessLog($str){
		QueueLog::businessLog($str);
	}

	public function errorLog($str){
		$this->error_msg = var_export($str, true);
		QueueLog::errorLog($str);
	}

	public function sqlErrorLog($str){
		$this->error_msg = var_export($str, true);
		QueueLog::sqlErrorLog($str);
	}
}
